package com.jdk.demo.nio.bilibili.Maodun.net.qiujuer.library.clink.impl.async;

import java.io.IOException;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import com.jdk.demo.nio.bilibili.Maodun.net.qiujuer.library.clink.core.IoArgs;
import com.jdk.demo.nio.bilibili.Maodun.net.qiujuer.library.clink.core.SendDispatcher;
import com.jdk.demo.nio.bilibili.Maodun.net.qiujuer.library.clink.core.SendPacket;
import com.jdk.demo.nio.bilibili.Maodun.net.qiujuer.library.clink.core.Sender;
import com.jdk.demo.nio.bilibili.Maodun.net.qiujuer.library.clink.utils.CloseUtils;

public class AsyncSendDispatcher implements SendDispatcher {

	private final AtomicBoolean isClosed =  new AtomicBoolean(false);
	private final Sender sender;
	private final Queue<SendPacket> queue = new ConcurrentLinkedQueue<>();
	private final AtomicBoolean isSending = new AtomicBoolean();

	private IoArgs ioArgs = new IoArgs();
	private SendPacket packetTemp;

	private int total; // 当前发送的packet的大小，以及进度
	private int position;

	public AsyncSendDispatcher(Sender sender) {
		this.sender = sender;
	}

	@Override
	public void send(SendPacket packet) {
		// packet入队列
		queue.offer(packet);
		if(isSending.compareAndSet(false, true)) {
			// 激活发送中
			sendNextPacket();
		}
	}

	private SendPacket takePacket() {
		SendPacket packet = queue.poll();
		if(packet != null && packet.isCanceled()) {
			// 已取消，不用发送，重新获取消息
			return takePacket();
		}
		return packet;
	}

	private void sendNextPacket() {
		SendPacket temp = packetTemp;
		if(temp != null) {
			CloseUtils.close(temp);
		}

		SendPacket packet = packetTemp = takePacket();
		if(packet == null) {
			// 队列为空，取消发送状态
			isSending.set(false);
			return;
		}

		total = packet.length();
		position  = 0;

		sendCurrentPacket();
	}

	private void sendCurrentPacket() {
		IoArgs args = ioArgs;
		// 开始、清理
		args.startWriting();

		if(position >= total) {
			// 一条完整的数据已经写完
			sendNextPacket();
			return;
		} else if(position == 0) {
			// 首包，需要携带长度信息
			args.writeLength(total);
		}

		byte[] bytes = packetTemp.bytes();
		// 把bytes的数据写入到IoArgs
		int count = args.readFrom(bytes, position);
		position += count;

		// 完成封装
		args.finishWriting();

		try {
			sender.sendAsync(args, ioArgsEventListener);
		} catch (IOException e) {
			closeAndNotify();
		}
	}

	@Override
	public void cancel(SendPacket packet) {
		IoArgs args = ioArgs;

		// 开始、清理
		args.startWriting();
		if(position >= total) {
			sendNextPacket();
			return;
		} else if(position == 0) {
			// 首包需要携带长度信息
			args.writeLength(total);
		}

		byte[] bytes = packetTemp.bytes();
		// 把bytes的数据写入到IoArgs
		int count = args.readFrom(bytes, position);
		position += count;

		// 完成封装
		args.finishWriting();

		try {
			sender.sendAsync(args, ioArgsEventListener);
		} catch (IOException e) {
			closeAndNotify();
		}
	}

	private void closeAndNotify() {
		CloseUtils.close(this);
	}

	private final IoArgs.IoArgsEventListener ioArgsEventListener = new IoArgs.IoArgsEventListener() {

		@Override
		public void onStarted(IoArgs args) {

		}

		@Override
		public void onCompleted(IoArgs args) {
			// 继续发送当前包
			sendCurrentPacket();
		}

	};

	@Override
	public void close() throws IOException {
		if(isClosed.compareAndSet(false, true)) {
			isSending.set(false);
			SendPacket packet = this.packetTemp;
			if(packet != null) {
				packetTemp = null;
				CloseUtils.close(packetTemp);
			}
		}
	}

}
